#pragma once

#include "../rpc.pb.h"
#include "../echo_service.pb.h"
#include "../RpcChannel.h"
#include "../../TcpConnection.h"
#include "../../TcpClient.h"
#include "../../TcpServer.h"
#include "../../EventLoop.h"
#include <memory>
#include "../RpcController.h"
#include "../../Channel.h"
#include "../../noncopyable.h"

// #include "MurmurHash3.h"
#include <stdio.h>
#include "ConsistentHash.h"
#include "../../util.h"

using namespace std::placeholders;

class BackendSession
{
public:
    BackendSession(EventLoop* loop, const string& ip, const int port, ConsistentHash* chs)
        : loop_(loop)
        , client_(loop, ip, port)
        , nextId_(0)
        , backendAddr_(ip + ":" + std::to_string(port))
        , isAvailable_(true)
        , consistentHashPtr_(chs)
    {
        client_.setConnectionCallback(std::bind(&BackendSession::onConnection, this, _1));
        client_.setMessageCallback(std::bind(&BackendSession::onMessage, this, _1, _2));
        client_.enableRetry();
    }

    std::string getBackendAddr() const { return backendAddr_; }

    void connect() { client_.connect(); }

    bool send(RpcMessage& msg, const TcpConnectionPtr& clientConn, std::string& key)
    {
        loop_->assertInLoopThread();
        std::lock_guard<std::mutex> lk(mutex_);
        if (conn_)
        {
            uint64_t id = ++nextId_;
            cout << getTid() << " unique id : " << id << endl;
            Request r = { msg.id(), clientConn };
            assert(outstandings_.find(id) == outstandings_.end());
            outstandings_[id] = r;
            msg.set_id(id);
            // string type = msg.type() == REQUEST ? "REQUEST" : msg.type() == RESPONSE ? "RESPONSE" : "NO_service";
            // cout << to_string(msg.id()) + " type: " + type + " message.service(): " + msg.service() + "\n";
            // std::cout << "type: " + type + " send to backsession\n";
            // std::cout << msg.DebugString();
            sendRpcMessage(conn_, msg);
            return true;
        }
        else
            return false;
    }


private:
    void onConnection(const TcpConnectionPtr& conn)
    {
        if (conn->connected())
        {
            conn_ = conn;
            // consistentHashPtr_->addNode(this);
            consistentHashPtr_->modify(this, 0);
        }
        else
        {
            conn_.reset();
            isAvailable_ = false;
            // consistentHashPtr_->removeNode(this);
            consistentHashPtr_->modify(this, 1);
        }
    }

    void onMessage(const TcpConnectionPtr& conn, Buffer* buf)
    {
        while (buf->readable()) {
            RpcMessagePtr rpcmessage( prototype_->New() );
            std::string msg;
            if (!parseMessage(conn, buf, msg)) {
                std::cout << "wait for more data\n";
                return ;
            }
            rpcmessage->ParseFromString(msg);

            onRpcMessage(conn, rpcmessage);
        }
    }

    void onRpcMessage(const TcpConnectionPtr& conn, const RpcMessagePtr& msg)
    {
        std::lock_guard<std::mutex> lk(mutex_);
        loop_->assertInLoopThread();
        std::map<uint64_t, Request>::iterator it = outstandings_.find(msg->id());
        if (it != outstandings_.end())
        {
            uint64_t origId = it->second.origId;
            TcpConnectionPtr clientConn = it->second.clientConn.lock();
            outstandings_.erase(it);

            if (clientConn)
            {
                cout << getTid() << " unique id : " << msg->id() << " oriId: " << origId << endl;
                msg->set_id(origId);
                // string type = msg->type() == REQUEST ? "REQUEST" : msg->type() == RESPONSE ? "RESPONSE" : "NO_service";
                // cout << to_string(msg->id()) + " type: " + type + " message.service(): " + msg->service() + "\n";
                // std::cout << "type: " + type + " send to client\n";
                // std::cout << msg->DebugString();
                sendRpcMessage(clientConn, *msg);
            }
        }
        else
        {
            cout << "not in outstandings_\n";
        }
    }

    struct Request
    {
        uint64_t origId;
        std::weak_ptr<TcpConnection> clientConn;
    };

    EventLoop* loop_;
    TcpClient client_;
    
    TcpConnectionPtr conn_;
    uint64_t nextId_;
    std::map<uint64_t, Request> outstandings_;
    
    const RpcMessagePtr prototype_;
    std::string backendAddr_;
    bool isAvailable_; // true: 可用; false: 不可用
    std::shared_ptr<ConsistentHash> consistentHashPtr_;
    std::mutex mutex_;
};